package com.fengkai.filepost.pdfoutstream.test.common.mq.translation;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.MessageSelector;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author Fengkai
 * @creat 2021/12/29
 */
public class SomeConsumer {
       public static void main(String[] args) throws MQClientException {
               //定义一个push型消费者
           DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("tcg");

           //指定nameserver
           consumer.setNamesrvAddr("192.168.157.129:9876");
           //指定从第一条开始消费
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
           //指定topic与tag
           consumer.subscribe("TranscationTopic", "*");

//           consumer.subscribe("myTopic", MessageSelector.bySql("age between 0 and 6"));  sql过滤 需要在broker启动时配置相应属性
//            //指定广播模式进行消费，默认为集群模式
//           consumer.setMessageModel(MessageModel.BROADCASTING);

           //注册消息监听器
           consumer.registerMessageListener(new MessageListenerConcurrently() {

               //一单broekr中有了其订阅的消息就会触发该方法的执行，其返回值为当前consumer消费的状态
               @Override
               public ConsumeConcurrentlyStatus consumeMessage (List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

                   //逐条消费消息
                   for (MessageExt messageExt : list) {
                       System.out.println(messageExt);
                   }


                   //返回消费状态：消费成功
                   return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
               }
           });

           //开启消费者
           consumer.start();
           System.out.println("consumer started");

       }
}
